/**
  <p>Title:TaskResourcePool  </p>
  <p>Description： 线程资源池</p>
  Package: com.travelsky.aic.data.jobs.agentmonitor.util.task
  Copyright: Copyright(c) 2020 Travelsky All Rights Reserved
  Company: Travelsky
  Software: AIC
  Date: 2021/1/15 9:33

  @author yanwen
 * @version 1.0.0
 */
package com.xiaoma.thread.pool;

import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListenableFutureTask;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author yanwen
 * @description 线程资源池
 * 现在满足基本的线程使用，自定义拒绝策略、队列、核心最大线程的定义，活跃期等设置
 * 后续会补充优先级，按照现在定时场景扩充任务归类、分组等实现
 */
public class TaskResourcePool implements ThreadFactory {
  private final Object poolSizeMonitor = new Object();
  /**
   * 核心线程数
   */
  private int corePoolSize = 1;
  /**
   * 最大线程数
   */
  private int maxPoolSize = 20;
  /**
   * 存活时间
   */
  private int keepAliveSeconds = 60;
  /**
   * 默认队列数
   */
  private int queueCapacity = 1024;
  /**
   * 是否允许核心线程超时
   */
  private boolean allowCoreThreadTimeOut = false;

  /**
   * 扩展装饰任务
   */
  private MonitorTaskDecorator taskDecorator;

  /**
   * 线程池执行器对象
   */
  private ThreadPoolExecutor threadPoolExecutor;

  /**
   * 封装任务属性容器
   */
  private final Map<Runnable, Object> decoratedTaskMap;

  /**
   * 默认拒绝策略
   * 可以自定义任务策略
   */
  private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();

  /**
   * 线程工厂
   */
  private final ThreadFactory threadFactory = this;

  /**
   * 初始化后的任务执行接口
   */
  private ExecutorService executor;

  private boolean waitForTasksToCompleteOnShutdown = false;

  private int awaitTerminationSeconds = 0;

  public TaskResourcePool() {
    this.decoratedTaskMap = new ConcurrentHashMap(16);
  }

  /**
   * 设置核心线程
   * @param corePoolSize 核心线程数量
   */
  public void setCorePoolSize(int corePoolSize) {
    synchronized(this.poolSizeMonitor) {
      this.corePoolSize = corePoolSize;
      if (this.threadPoolExecutor != null) {
        this.threadPoolExecutor.setCorePoolSize(corePoolSize);
      }

    }
  }

  /**
   * 获取核心线程数量
   * @return corePoolSize 核心线程数量
   */
  public int getCorePoolSize() {
    synchronized(this.poolSizeMonitor) {
      return this.corePoolSize;
    }
  }

  /**
   * 设置最大线程池大小
   * @param maxPoolSize 最大线程数量
   */
  public void setMaxPoolSize(int maxPoolSize) {
    synchronized(this.poolSizeMonitor) {
      this.maxPoolSize = maxPoolSize;
      if (this.threadPoolExecutor != null) {
        this.threadPoolExecutor.setMaximumPoolSize(maxPoolSize);
      }

    }
  }

  /**
   * 获取最大线程池大小
   * @return maxPoolSize 最大线程数量
   */
  public int getMaxPoolSize() {
    synchronized(this.poolSizeMonitor) {
      return this.maxPoolSize;
    }
  }

  /**
   * 设置线程活跃期
   * @param keepAliveSeconds 保持活跃时间
   */
  public void setKeepAliveSeconds(int keepAliveSeconds) {
    synchronized(this.poolSizeMonitor) {
      this.keepAliveSeconds = keepAliveSeconds;
      if (this.threadPoolExecutor != null) {
        this.threadPoolExecutor.setKeepAliveTime((long)keepAliveSeconds, TimeUnit.SECONDS);
      }

    }
  }

  /**
   * 获取活跃期设置
   * @return keepAliveSeconds 保持活跃时间
   */
  public int getKeepAliveSeconds() {
    synchronized(this.poolSizeMonitor) {
      return this.keepAliveSeconds;
    }
  }

  /**
   * 设置队列容量
   * @param queueCapacity 队列容量
   */
  public void setQueueCapacity(int queueCapacity) {
    this.queueCapacity = queueCapacity;
  }

  /**
   * 返回队列容量
   * @param allowCoreThreadTimeOut 超时时间
   */
  public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
    this.allowCoreThreadTimeOut = allowCoreThreadTimeOut;
  }

  /**
   * 任务修饰器
   * @param taskDecorator 任务修饰器
   */
  public void setTaskDecorator(MonitorTaskDecorator taskDecorator) {
    this.taskDecorator = taskDecorator;
  }

  /**
   * 设置拒绝策略
   * @param rejectedExecutionHandler 拒绝策略
   */
  public void setRejectedExecutionHandler(RejectedExecutionHandler rejectedExecutionHandler) {
    this.rejectedExecutionHandler =
        (rejectedExecutionHandler != null ? rejectedExecutionHandler : new ThreadPoolExecutor.AbortPolicy());
  }

  /**
   * 初始化线程池设置
   * 请勿删除其中注释代码
   */
  public void initialize() {
//    if (logger.isInfoEnabled()) {
//      logger.info("Initializing ExecutorService" + (this.beanName != null ? " '" + this.beanName + "'" : ""));
//    }
//    if (!this.threadNamePrefixSet && this.beanName != null) {
//      setThreadNamePrefix(this.beanName + "-");
//    }
    this.executor = initializeExecutor(this.threadFactory, this.rejectedExecutionHandler);
  }

  /**
   * 初始化线程池执行对象
   * @param threadFactory
   * @param rejectedExecutionHandler
   * @return ExecutorService 执行器业务对象
   */
  protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
    BlockingQueue<Runnable> queue = this.createQueue(this.queueCapacity);
    ThreadPoolExecutor executor;
    if (this.taskDecorator != null) {
      executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, (long)this.keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler) {
        @Override
        public void execute(Runnable command) {
          Runnable decorated = TaskResourcePool.this.taskDecorator.decorate(command);
          if (decorated != command) {
            TaskResourcePool.this.decoratedTaskMap.put(decorated, command);
          }

          super.execute(decorated);
        }
      };
    } else {
      executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, (long)this.keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler);
    }

    if (this.allowCoreThreadTimeOut) {
      executor.allowCoreThreadTimeOut(true);
    }

    this.threadPoolExecutor = executor;
    return executor;
  }

  /**
   * 创建队列
   * @param queueCapacity 队列容量
   * @return BlockingQueue<Runnable> 固定得队列
   */
  protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
    return (BlockingQueue)(queueCapacity > 0 ? new LinkedBlockingQueue(queueCapacity) : new SynchronousQueue());
  }

  /**
   * 线程池中获取线程执行对象
   * @return ThreadPoolExecutor 线程资源执行器
   * @throws IllegalStateException 非法状态异常
   */
  public ThreadPoolExecutor getThreadPoolExecutor() throws IllegalStateException {
    //Assert.state(this.threadPoolExecutor != null, "ThreadPoolTaskExecutor not initialized");
    return this.threadPoolExecutor;
  }

  /**
   * 获取线程池大小
   * @return corePoolSize || poolSize
   */
  public int getPoolSize() {
    return this.threadPoolExecutor == null ? this.corePoolSize : this.threadPoolExecutor.getPoolSize();
  }

  /**
   * 获取活跃线程数
   * @return activeCount 活跃线程数
   */
  public int getActiveCount() {
    return this.threadPoolExecutor == null ? 0 : this.threadPoolExecutor.getActiveCount();
  }

  /**
   * 执行任务
   * @param task 线程任务
   */
  public void execute(Runnable task) {
    ThreadPoolExecutor executor = this.getThreadPoolExecutor();

    try {
      executor.execute(task);
    } catch (RejectedExecutionException var4) {
      throw new MonitorRejectedException("Executor [" + executor + "] did not accept task: " + task, var4);
    }
  }

  /**
   * 执行任务2
   * @param task
   * @param startTimeout 超时时间
   */
  public void execute(Runnable task, long startTimeout) {
    this.execute(task);
  }

  /**
   * 执行有返回值的任务 Runnable
   * @param task
   * @return
   */
  public Future<?> submit(Runnable task) {
    ThreadPoolExecutor executor = this.getThreadPoolExecutor();

    try {
      return executor.submit(task);
    } catch (RejectedExecutionException var4) {
      throw new MonitorRejectedException("Executor [" + executor + "] did not accept task: " + task, var4);
    }
  }

  /**
   * 执行有返回值的任务 Callable
   * @param task
   * @param <T>
   * @return
   */
  public <T> Future<T> submit(Callable<T> task) {
    ThreadPoolExecutor executor = this.getThreadPoolExecutor();

    try {
      return executor.submit(task);
    } catch (RejectedExecutionException var4) {
      throw new MonitorRejectedException("Executor [" + executor + "] did not accept task: " + task, var4);
    }
  }

  /**
   * 含监听有偿任务 Runnable
   * @param task
   * @return
   */
  public ListenableFuture<?> submitListenable(Runnable task,Object result) {
    ThreadPoolExecutor executor = this.getThreadPoolExecutor();

    try {
      ListenableFutureTask<?> future = ListenableFutureTask.create(task, result);
          executor.execute(future);
      return future;
    } catch (RejectedExecutionException var4) {
      throw new MonitorRejectedException("Executor [" + executor + "] did not accept task: " + task, var4);
    }
  }

  /**
   * 含监听有偿任务 Callable
   * @param task
   * @param <T>
   * @return
   */
  public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
    ThreadPoolExecutor executor = this.getThreadPoolExecutor();

    try {
      ListenableFutureTask<T> future = ListenableFutureTask.create(task);
      executor.execute(future);
      return future;
    } catch (RejectedExecutionException var4) {
      throw new MonitorRejectedException("Executor [" + executor + "] did not accept task: " + task, var4);
    }
  }

  /**
   * 取消还在运行得任务
   * @param task 线程
   */
  protected void cancelRemainingTask(Runnable task) {
    if (task instanceof Future) {
      ((Future<?>) task).cancel(true);
    }
    Object original = this.decoratedTaskMap.get(task);
    if (original instanceof Future) {
      ((Future)original).cancel(true);
    }

  }

  /**
   *
   * @param runnable
   * @return Thread 创建线程
   */
  @Override
  public Thread newThread(Runnable runnable) {
    return createThread(runnable);
  }

  /**
   * 创建线程
   * 后续实现分组、优先级、线程名称等
   * 不要删除注释类
   * @param runnable
   * @return Thread
   */
  public Thread createThread(Runnable runnable) {
    //Thread thread = new Thread(getThreadGroup(), runnable, nextThreadName());
    Thread thread = new Thread(runnable);
//    thread.setPriority(getThreadPriority());
//    thread.setDaemon(isDaemon());
    return thread;
  }

  /**
   * 关闭执行器
   */
  public void shutdown() {
    //待补充日志等信息
    if (this.executor != null) {
      if (this.waitForTasksToCompleteOnShutdown) {
        this.executor.shutdown();
      }
      else {
        for (Runnable remainingTask : this.executor.shutdownNow()) {
          cancelRemainingTask(remainingTask);
        }
      }
      awaitTerminationIfNecessary(this.executor);
    }
  }

  /**
   * 等待一个执行器执行结束
   * @param executor 执行对象
   */
  private void awaitTerminationIfNecessary(ExecutorService executor) {
    if (this.awaitTerminationSeconds > 0) {
      try {
        if (!executor.awaitTermination(this.awaitTerminationSeconds, TimeUnit.SECONDS)) {
//           log.warn("Timed out while waiting for executor to terminate");
        }
      }
      catch (InterruptedException ex) {
//        log.warn("Timed out while waiting for executor to terminate");
        Thread.currentThread().interrupt();
      }
    }
  }

}
